package com.wd.mqtt;

import org.smartboot.mqtt.client.MqttClient;
import org.smartboot.socket.StateMachineEnum;
import org.smartboot.socket.buffer.BufferPagePool;
import org.smartboot.socket.enhance.EnhanceAsynchronousChannelProvider;
import org.smartboot.socket.extension.plugins.MonitorPlugin;
import org.smartboot.socket.extension.processor.AbstractMessageProcessor;
import org.smartboot.socket.extension.protocol.StringProtocol;
import org.smartboot.socket.transport.AioQuickClient;
import org.smartboot.socket.transport.AioSession;

import java.io.IOException;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.spi.AsynchronousChannelProvider;
import java.util.concurrent.ThreadFactory;

public class SmqttImpl implements MqttClientApi {
    private String address;

    private MqttClient mqttClient;

    public static void init() throws IOException {
        AsynchronousChannelProvider provider = new EnhanceAsynchronousChannelProvider(true);
        AsynchronousChannelGroup[] groups = new AsynchronousChannelGroup[1];
        for (int i = 0; i < groups.length; i++) {
            groups[i] = provider.openAsynchronousChannelGroup(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r);
                }
            });
        }
        AbstractMessageProcessor<String> processor = new AbstractMessageProcessor<String>() {
            @Override
            public void process0(AioSession session, String msg) {

            }

            @Override
            public void stateEvent0(AioSession session, StateMachineEnum stateMachineEnum, Throwable throwable) {

            }
        };
        processor.addPlugin(new MonitorPlugin<>(5));
//        BufferPagePool bufferPagePool = new BufferPagePool(1024 * 1024, Runtime.getRuntime().availableProcessors() + 1, true);
//        for (int i = 0; i < count; i++) {
//            AioQuickClient client = new AioQuickClient(host, port, new StringProtocol(), processor);
//            client.setReadBufferSize(1024).setBufferFactory(() -> bufferPagePool);
//            try {
//                client.start(groups[i % groups.length]);
//                synchronized (client) {
//                    client.wait(1);
//                }
//            } catch (Throwable throwable) {
//                count--;
//                throwable.printStackTrace();
//                client.shutdownNow();
//            }
//        }
    }

    public SmqttImpl(String address) {
        this.address = address;
    }

    @Override
    public void conect(String clientId) {
        mqttClient = new MqttClient(address,clientId);

        mqttClient.connect();
    }

    @Override
    public void push() {

    }
}
